package com.bbx.flink.demo.evictors;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.evictors.DeltaEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import sun.awt.SunHints;

public class DeltaEvictorDemo {

    /**
     * DeltaEvictor
     * 窗口内最后一个元素和窗口内每一个元素（包含最后一个）进行函数运算，运算结果与设定的阈值进行比较，移除运算结果大于、等于 阈值的元素
     * @param args
     * @throws Exception
     */
    public static void main(String [] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.10.131" ,10003)
                .map((MapFunction<String ,Integer>) value->Integer.parseInt(value))
                .keyBy(i->i%2)
                .timeWindow(Time.minutes(1))
                // DeltaEvictor<T, W> of(double threshold, DeltaFunction<T> deltaFunction)
                //threshold: 设定的阈值
                //DeltaFunction<T> deltaFunction   窗口内每个元素和窗口内最后一个元素的运算结果
                //如果运算结果大于或者 等于 阈值，则将该元素舍去（该元素是指正在源最后一个元素运算的元素）
                //例如：下面示例 ：
                // 阈值设定为30
                // 窗口内输入元素为 1 ，3，5，11，33，55  最后元素为55 ，
                // 运算函数为  新值 -旧值（55-每一个元素）
                //则 55-1=54 >30  舍去元素1
                //则 55-3=52 >30  舍去元素3
                //则 55-5=50 >30  舍去元素5
                //则 55-11=44 >30  舍去元素11
                //则 55-33=22 < 30 保留元素33
                //则 55-55=0 >30  保留元素55
                //因此 最后计算结果为 33+55= 88
                .evictor(DeltaEvictor.of(30,(oldData,newData)->{return newData- oldData;}))
                .sum(0)
                .print();
        env.execute();
    }


}



